package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

public class Demo9INdex {

    private HConnection connection;

    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {
        // 1、创建连接

        Configuration configuration = new Configuration();

        /*
         * 不需要执行hmaster和regionserver的地址m  hclient会自动去zookeeper中获取，
         * 由于获取到的是一个域名，所以本地必须配置hosts
         *
         */

        //设置zookeeper地址
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        try {

            //创建连接  对数据增删改查使用
            connection = HConnectionManager.createConnection(configuration);

            /*
             * 创建修改删除表使用
             *
             */
            hBaseAdmin = new HBaseAdmin(connection);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void putAll() throws Exception {


        /**
         * create 'bigdata:student1','info'
         *
         */

        HTableInterface table = connection.getTable(TableName.valueOf("bigdata:student1"));

        BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));

        String line;
        while ((line = bufferedReader.readLine()) != null) {
            String[] split = line.split(",");
            String id = split[0];
            String name = split[1];
            String age = split[2];
            String gender = split[3];
            String clazz = split[4];

            ArrayList<Put> puts = new ArrayList<>();


            for (int i = 0; i < 1000; i++) {

                String newId = id + i;


                //将列值放到rowkey中实现二级索引
                String rk = gender + "_" + newId;

                Put put = new Put(rk.getBytes());
                put.add("info".getBytes(), "name".getBytes(), name.getBytes());
                put.add("info".getBytes(), "age".getBytes(), age.getBytes());
                put.add("info".getBytes(), "clazz".getBytes(), clazz.getBytes());


                puts.add(put);
            }

            //批量插入
            table.put(puts);
        }
    }


    @Test
    public void getRoekeyFilter() throws Exception {

        long start = System.currentTimeMillis();
        /**
         * 查询新别为男的数据
         *
         */

        HTableInterface table = connection.getTable(TableName.valueOf("bigdata:student1"));

        Scan scan = new Scan();


        //前缀比较器
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("男*".getBytes());

        //行键过滤器
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

        //增加过滤器
        scan.setFilter(rowFilter);

        ResultScanner scanner = table.getScanner(scan);

        //获取结果
        print(scanner);

        long end = System.currentTimeMillis();
        System.out.println(end - start);


    }


    public static void print(ResultScanner scanner) throws IOException {
        Result next;

        while ((next = scanner.next()) != null) {

            String gender = Bytes.toString(next.getValue("info".getBytes(), "gender".getBytes()));
            String name = Bytes.toString(next.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(next.getValue("info".getBytes(), "age".getBytes()));
            String clazz = Bytes.toString(next.getValue("info".getBytes(), "clazz".getBytes()));
            String id = Bytes.toString(next.getRow());

            //System.out.println(id + "\t" + name + "\t" + age + '\t' + gender + '\t' + clazz);
        }
    }


}
